package cn.neo.hhmq.client;

import cn.neo.hhmq.model.Message;
import lombok.Getter;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * 消费者
 */
public class HHConsumer<T> {

    HHBroker broker;

    String topic;

    HHMq mq;

    private String id;


    static AtomicInteger idgen = new AtomicInteger(0);

    public HHConsumer(HHBroker broker) {
        this.broker = broker;
        this.id = "CID" + idgen.getAndIncrement();
    }

    // 订阅
    public void sub(String topic) {
        broker.sub(topic, id);
    }

    public void unsub(String topic) {
        broker.unsub(topic, id);
    }

    public Message<?> recv(String topic) {
        return broker.recv(topic, id);
    }

    public boolean ack(String topic, int offset) {
        return broker.ack(topic, id, offset);
    }

    public boolean ack(String topic, Message<?> message) {
        int offset = Integer.parseInt(message.getHeaders().get("X-offset"));
        return ack(topic, offset);
    }

    // 被动: 添加监听者---发布订阅
    public void listen(String topic, HHListener<T> listener) {
        this.listener = listener;
        broker.addConsumer(topic,this);
    }

    @Getter
    private HHListener<T> listener;
}
